package com.abyss.sse;

import com.abyss.avro.SseAvro;
import com.abyss.util.DateTimeUtil;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.flume.Context;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.event.EventHelper;
import org.apache.flume.serialization.BodyTextEventSerializer;
import org.apache.flume.serialization.EventDeserializer;
import org.apache.flume.source.AbstractSource;

import java.beans.EventHandler;
import java.io.*;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Date 2020/9/13
 */
//1.实现自定义source接口
public class SseQuotSource extends AbstractSource implements PollableSource, Configurable {
    private String host;
    private Integer port;
    private String userName;
    private String password;
    private String ftpDirectory;
    private String fileName;
    private String localDirectory;
    private Integer corePoolSize;
    private Integer maxPoolSize;
    private Integer keepAliveTime;
    private Integer capacity;
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 实现步骤
     * 1.实现自定义source接口
     * 2.初始化参数（初始化source参数和线程池）
     * 3.判断是否是交易时间段
     * 4.异步处理
     * 5.设置延时时间
     */

    @Override
    public Status process() {

        //3.判断是否是交易时间段
        //获取当前时间
//        long time = new Date().getTime();
//        if (time < DateTimeUtil.closeTime && time > DateTimeUtil.openTime) {

            //4.异步处理
            threadPoolExecutor.execute(new AsyncTask());

            try {
                Thread.sleep(1000L); //睡眠1s
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Status.READY; //有数据
//        }

//        return Status.BACKOFF;//无数据
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }



    //2.初始化参数（初始化source参数和线程池）
    @Override
    public void configure(Context context) {

        host = context.getString("host");
        port = context.getInteger("port");
        userName = context.getString("userName");
        password = context.getString("password");
        ftpDirectory = context.getString("ftpDirectory");
        fileName = context.getString("fileName");
        localDirectory = context.getString("localDirectory");
        corePoolSize = context.getInteger("corePoolSize");
        maxPoolSize = context.getInteger("maxPoolSize");
        keepAliveTime = context.getInteger("keepAliveTime");
        capacity = context.getInteger("capacity");

        //创建线程池
        threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
                keepAliveTime, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(capacity));
    }
    /**
     * 创建异步线程，执行业务逻辑
     */
    //1.创建异步线程task
    private class AsyncTask implements Runnable {

        @Override
        public void run() {
            /**
             * 开发步骤：
             * 1.创建异步线程task
             * 2.下载行情文件
             * 3.解析并发送数据
             *   数据转换成avro
             *   数据序列化
             * 4.发送数据到channel
             */
            //2.下载行情文件（了解）
            download();
            //3.解析并发送数据
            try {
                //读取本地文件
                BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(new File(localDirectory + "/" + fileName))));
                String str;
                //轮询文本文件
                int i = 0;
                while ((str = br.readLine()) != null) {
                    String[] arr = str.split("\\|");
                    if (i == 0) {
                        if (arr[8].trim().startsWith("E")) {
                            break;
                        }
                    } else {
                        //数据转换成avro
                        SseAvro sseAvro = transfer(arr);
                        //数据序列化
                        byte[] bte = serializer(sseAvro);
                        // 4.发送数据到channel
                        getChannelProcessor().processEvent(EventBuilder.withBody(bte));
                    }
                    i++;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        //序列化操作
        private byte[] serializer(SseAvro sseAvro) {

            SpecificDatumWriter<Object> datumWriter = new SpecificDatumWriter<>(sseAvro.getSchema());

            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            //编码成二进制序列化对象
            BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(bos, null);
            try {
                datumWriter.write(sseAvro, binaryEncoder);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return bos.toByteArray();
        }

        //字符串数据转换成avro对象
        private SseAvro transfer(String[] arr) {
            SseAvro sseAvro = new SseAvro();
            sseAvro.setMdStreamID(arr[0].trim());
            sseAvro.setSecurityID(arr[1].trim());
            sseAvro.setSymbol(arr[2].trim());
            sseAvro.setTradeVolume(Long.valueOf(arr[3].trim()));
            sseAvro.setTotalValueTraded(Long.valueOf(arr[4].trim()));
            sseAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
            sseAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
            sseAvro.setHighPrice(Double.valueOf(arr[7].trim()));
            sseAvro.setLowPrice(Double.valueOf(arr[8].trim()));
            sseAvro.setTradePrice(Double.valueOf(arr[9].trim()));
            sseAvro.setClosePx(Double.valueOf(arr[10].trim()));
            sseAvro.setTradingPhaseCode("T01");
            sseAvro.setTimestamp(new Date().getTime());
            return sseAvro;
        }

        //下载ftp文件
        private void download() {

            /**
             * 开发步骤：
             * 1.初始化ftp连接
             * （1）设置IP和Port
             * （2）设置登陆用户名和密码
             * (3) 设置编码格式
             * （4）判断是否连接成功（FTPReply）
             * 2.切换工作目录，设置被动模式
             * 3.获取工作目录的文件信息列表
             * 4.输出文件
             * 5.退出，返回成功状态
             */

            FTPClient ftpClient = new FTPClient();
            try {
                //（1）设置IP和Port
                ftpClient.connect(host, port);
                //（2）设置登陆用户名和密码
                ftpClient.login(userName, password);
                //(3) 设置编码格式
                ftpClient.setControlEncoding("UTF-8");
                //（4）判断是否连接成功（FTPReply）
                //返回码
                int replyCode = ftpClient.getReplyCode();
                if (FTPReply.isPositiveCompletion(replyCode)) {
                    //2.切换工作目录，设置被动模式
                    ftpClient.changeWorkingDirectory(ftpDirectory);
                    //设置被动模式,主要使用被动模式
                    ftpClient.enterLocalPassiveMode();
                    //禁用服务端远程验证，会验证提交得ip与主机ip是否一致，如果设置为true，ip不一致得时候，就报错
                    ftpClient.setRemoteVerificationEnabled(false);
                    //3.获取工作目录的文件信息列表
                    FTPFile[] ftpFiles = ftpClient.listFiles();
                    for (FTPFile ftpFile : ftpFiles) {
                        //4.输出文件
                        ftpClient.retrieveFile(ftpFile.getName(), new FileOutputStream(new File(localDirectory + "/" + fileName)));
                    }
                }

                //5.退出，返回成功状态
                ftpClient.logout();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
